package org.datastreamapi.transform

import org.FlinkStreamApp
import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * description ：转换-flatmap
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/21 19:39
 * modified By ：
 * version:    : 1.0
 */
object FlagMap extends FlinkStreamApp {

  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 创建数据
    val source = environment.fromElements("1 2", "3 4", "5 6")

    // 转换 匿名函数
    //source.flatMap(ws=> ws.split("\\s")).print

    // 转换 接口方式
    source.flatMap(new MyFlatMapFunction).print()
  }

  class MyFlatMapFunction extends FlatMapFunction[String, String] {
    override def flatMap(value: String, out: Collector[String]) = {
      val splits: Array[String] = value.split("\\s")
      for (elem <- splits) {
        out.collect(elem)
      }
    }
  }

}
